package gateway

import (
	"context"
	"errors"
	"fmt"
	"github.com/spf13/cast"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"io"
	"log"
	"server/component"
	"server/im/entity"
	"server/im/protocol"
	pb "server/im/rpc"
	"server/repository/model"
	"sync"
	"time"
)

type BaseGateway struct {
	GatewayUni        string
	ClientMap         map[string]*ClientItem
	ClientChan        chan entity.GatewayClientEvent
	Config            component.ImServerConfig
	CoreInputChan     chan entity.CoreInModel
	CoreOutputChan    chan entity.CoreOutModel
	GatewayInputChan  chan entity.GatewayInModel
	GatewayOutputChan chan entity.GatewayOutModel
	BroadcastChan     chan entity.GatewayBroadcastModel
	Fn                func()
	FnStartListen     func()
	FnStopListen      func()
	rpcClient         pb.HubClient
	rpcSteam          *grpc.BidiStreamingClient[pb.CoreInModel, pb.CoreOutModel]
	ctx               context.Context
	innerWG           sync.WaitGroup
	exitChan          chan struct{}
}

func (o *BaseGateway) GetUniId() string {
	return o.GatewayUni
}

func (o *BaseGateway) GetClientChan() chan entity.GatewayClientEvent {
	return o.ClientChan
}

func (o *BaseGateway) GetCoreInputChan() chan entity.CoreInModel {
	return o.CoreInputChan
}

func (o *BaseGateway) GetCoreOutputChan() chan entity.CoreOutModel {
	return o.CoreOutputChan
}

func (o *BaseGateway) GetGatewayInputChan() chan entity.GatewayInModel {
	return o.GatewayInputChan
}

func (o *BaseGateway) GetGatewayOutputChan() chan entity.GatewayOutModel {
	return o.GatewayOutputChan
}

func (o *BaseGateway) Start(wg *sync.WaitGroup) {
	defer func() {
		fmt.Println("wg " + o.GatewayUni + " done")
		wg.Done()
	}()
	fmt.Println("gateway" + o.GatewayUni + "start")
	o.rpcLinked()

	o.innerWG.Add(5)

	go o.sendToHub()
	go o.receiveFromHub()
	go o.broadcast()
	go o.onlineListen()
	go o.startListen()

	o.innerWG.Wait()
}

func (o *BaseGateway) Stop() {
	(*o.rpcSteam).CloseSend()
	close(o.exitChan)
	for _, item := range o.ClientMap {
		item.conn.CloseFn()
	}
	o.FnStopListen()
}

// 网关监听
func (o *BaseGateway) Conn2Client(conn GateWayConn) {
	client := newClientItem(nil, conn, o)
	go client.Work()

	rand := component.RandStr(32)
	tmpUniId := cast.ToString(time.Now().Unix()) + "_" + rand
	client.UserUniId = tmpUniId

	o.ClientChan <- entity.GatewayClientEvent{
		UserUni:   tmpUniId,
		Data:      &client,
		EventType: entity.ClientEventConnect,
	}
}

// 在线变化
func (o *BaseGateway) onlineListen() {
	defer func() {
		o.innerWG.Done()
	}()
	for {
		select {
		case <-o.exitChan:
			return
		case ClientEvent := <-o.ClientChan:
			fmt.Printf("cline chan change %+v \n", ClientEvent)
			//连接
			if ClientEvent.EventType == entity.ClientEventConnect {
				client, ok := ClientEvent.Data.(*ClientItem)
				if !ok {
					component.Log.Error("connect client err", ClientEvent.Data)
					continue
				}
				o.ClientMap[ClientEvent.UserUni] = client
			}
			//登录
			if ClientEvent.EventType == entity.ClientEventLogin {
				client, ok := ClientEvent.Data.(*ClientItem)
				if !ok {
					component.Log.Error("connect client err", ClientEvent.Data)
					continue
				}
				fmt.Println("client", client)
				o.ClientMap[client.UserUniId] = client
				delete(o.ClientMap, ClientEvent.UserUni) //todo lock
			}
			//断开
			if ClientEvent.EventType == entity.ClientEventLostConnect {
				client, ok := o.ClientMap[ClientEvent.UserUni]
				if ok {
					//停止client
					client.Stop()

					delete(o.ClientMap, ClientEvent.UserUni)

					//上报core断开
					ctx := metadata.NewOutgoingContext(
						context.Background(),
						metadata.Pairs(
							"authorization", "Bearer valid_token_12345", // 这里的 Token 必须是有效的
							"gateway", o.GetUniId(),
						))
					offlineReq := pb.CoreInModel{
						GatewayInModel: &pb.GatewayInModel{
							UserUni: ClientEvent.UserUni,
						},
						GatewayId: o.GatewayUni,
						MsgType:   entity.MsgTypeNotify,
					}
					o.rpcClient.ChatOffline(ctx, &offlineReq)
				}
			}

			fmt.Printf("cur gateway online %+v", o.ClientMap)
		}
	}
}

// 网关监听
func (o *BaseGateway) startListen() {
	defer func() {
		o.innerWG.Done()
	}()
	o.FnStartListen()
}

// 广播
func (o *BaseGateway) broadcast() {
	defer func() {
		o.innerWG.Done()
	}()
	for {
		select {
		case <-o.exitChan:
			return
		case broadcastModel := <-o.BroadcastChan:
			online := o.ClientMap
			if broadcastModel.SendType == entity.BroadcastAll {
				for _, client := range online {
					client.WriteChan <- broadcastModel.GatewayOutModel
				}
			} else if broadcastModel.SendType == entity.BroadcastInclude {
				for uniId, client := range online {
					if component.SliceIndex(broadcastModel.ReceiverInclude, uniId) >= 0 {
						client.WriteChan <- broadcastModel.GatewayOutModel
					}
				}
			} else if broadcastModel.SendType == entity.BroadcastExclude {
				for uniId, client := range online {
					if component.SliceIndex(broadcastModel.ReceiverExclude, uniId) == -1 {
						client.WriteChan <- broadcastModel.GatewayOutModel
					}
				}
			}
		}
	}
}

func (o *BaseGateway) rpcLinked() {
	o.rpcClientBuild()
	o.rpcChatBuild()
}

func (o *BaseGateway) rpcClientBuild() {
	rpcPort := cast.ToString(component.ConfigModel.RpcPort)
	conn, err := grpc.NewClient(":"+rpcPort,
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	o.rpcClient = pb.NewHubClient(conn)
}

func (o *BaseGateway) rpcChatBuild() {

	ctx := metadata.NewOutgoingContext(
		context.Background(),
		metadata.Pairs(
			"authorization", "Bearer valid_token_12345", // 这里的 Token 必须是有效的
			"gateway", o.GetUniId(),
		))

	stream, err := o.rpcClient.Chat(ctx)

	if err != nil {
		log.Fatalf("client.RouteChat failed: %v", err)
	}

	fmt.Println("gateway" + o.GatewayUni + " hub rpc linked")

	o.rpcSteam = &stream
}

func (o *BaseGateway) sendToHub() {
	defer func() {
		o.innerWG.Done()
	}()
	for {
		select {
		case <-o.exitChan:
			return
		//收到请求信息 发送hub处理
		case clientMsg := <-o.GatewayInputChan:

			msgBytes := clientMsg.Data

			if clientMsg.Cmd == protocol.ReqLogin {
				userUni := clientMsg.UserUni
				clientItem, ok := o.ClientMap[userUni]
				if !ok {
					component.Log.Error(errors.New("client not exist" + userUni))
					return
				}

				if clientItem.user == nil {

					//连接登录
					ctx := metadata.NewOutgoingContext(
						context.Background(),
						metadata.Pairs(
							"authorization", "Bearer valid_token_12345", // 这里的 Token 必须是有效的
							"gateway", o.GetUniId(),
						))
					loginReq := pb.CoreInModel{
						GatewayInModel: &pb.GatewayInModel{
							Data:    msgBytes,
							Cmd:     uint32(clientMsg.Cmd),
							Ip:      clientMsg.Ip,
							User:    nil,
							UserUni: clientMsg.UserUni,
						},
						GatewayId: o.GatewayUni,
						MsgType:   entity.MsgTypeNotify,
					}
					coreOut, err := o.rpcClient.ChatLogin(ctx, &loginReq)

					rt := map[string]any{
						"code": 101,
						"msg":  "fail",
					}

					if err != nil { //fail
						rt = map[string]any{
							"code": 102,
							"msg":  err.Error(),
						}
					} else { //success
						var userModel model.UserModel
						err := component.JsonDecode(coreOut.GatewayOutModel.Data, &userModel)
						if err != nil {
							component.Log.Error(err)
							rt = map[string]any{
								"code": 103,
								"msg":  err.Error(),
							}
							return
						} else {
							//login success event
							clientItem.UserUniId = userModel.Uuid
							clientItem.user = &entity.UserModel{
								Id:    userModel.Id,
								Name:  userModel.Name,
								Token: userModel.Token,
								Uuid:  userModel.Uuid,
								Ip:    clientMsg.Ip,
							}
							o.ClientChan <- entity.GatewayClientEvent{
								UserUni:   userUni,
								Data:      clientItem,
								EventType: entity.ClientEventLogin,
							}

							rt = map[string]any{
								"code": 0,
								"msg":  "success",
							}
						}
					}

					rtBytes, _ := component.JsonEncode(rt)
					gatewayMsg := entity.GatewayOutModel{
						Data: rtBytes,
						Cmd:  uint16(protocol.RespLogin),
					}
					clientItem.WriteChan <- gatewayMsg

				}

			} else {
				//其他消息
				tmp := &pb.CoreInModel{
					GatewayInModel: &pb.GatewayInModel{
						Data: msgBytes,
						Cmd:  uint32(clientMsg.Cmd),
						Ip:   clientMsg.Ip,
						User: &pb.UserModel{
							Id:    int32(clientMsg.User.Id),
							Name:  clientMsg.User.Name,
							Token: clientMsg.User.Token,
							Uuid:  clientMsg.User.Uuid,
							Ip:    clientMsg.User.Ip,
						},
						UserUni: clientMsg.UserUni,
					},
					GatewayId: o.GatewayUni,
					MsgType:   entity.MsgTypeNormal,
				}
				fmt.Println("send", tmp)
				(*o.rpcSteam).Send(tmp)
			}
		}
	}
}

func (o *BaseGateway) receiveFromHub() {
	defer func() {
		o.innerWG.Done()
	}()
	for {
		coreMsg, err := (*o.rpcSteam).Recv()
		if err != nil {
			if err != io.EOF {
				component.Log.Error(err)
			}
			return
		}
		log.Printf("Got message %+v ", coreMsg)

		if coreMsg.MsgType == entity.MsgTypeNormal {
			gatewayMsg := entity.GatewayOutModel{
				Data: coreMsg.GatewayOutModel.Data,
				Cmd:  uint16(coreMsg.GatewayOutModel.Cmd),
			}

			switch coreMsg.ReceiverType {
			case entity.ReceiverTypeSingle:
				online := o.ClientMap
				uniId := coreMsg.Clients[0]
				if client, ok := online[uniId]; ok {
					client.WriteChan <- gatewayMsg
				}
			case entity.ReceiverTypeBroadcast:
				o.BroadcastChan <- entity.GatewayBroadcastModel{
					GatewayOutModel: gatewayMsg,
					SendType:        entity.BroadcastAll,
				}
			case entity.ReceiverTypeGroup:
				o.BroadcastChan <- entity.GatewayBroadcastModel{
					GatewayOutModel: gatewayMsg,
					SendType:        entity.BroadcastInclude,
					ReceiverInclude: coreMsg.Clients,
				}
			case entity.ReceiverTypeExclude:
				o.BroadcastChan <- entity.GatewayBroadcastModel{
					GatewayOutModel: gatewayMsg,
					SendType:        entity.BroadcastExclude,
					ReceiverExclude: coreMsg.Clients,
				}
			}
		}
	}

}
